package com.lefu.risk.storm.bolt;

import com.lefu.risk.storm.common.constants.Constant;
import com.lefu.risk.storm.utils.JedisPoolUtil;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by mwang on 2016/12/29.
 */
public class ResultDataBolt extends BaseRichBolt {


    private static Map<String,Long> errorCodeMap = null;
    private OutputCollector collector;

    private JedisPoolUtil jedisPoolUtil = null;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        errorCodeMap = new HashMap<String,Long>();
        jedisPoolUtil = new JedisPoolUtil(Constant.REDIS_PROPERTIES_NAME);
    }

    @Override
    public void execute(Tuple input) {
        try {
            long resultDataBoltTime = Long.parseLong(jedisPoolUtil.time().get(0)) * Constant.SECOND_TO_MICROSECONDS + Long.parseLong(jedisPoolUtil.time().get(1));
            String[] strArr = input.getString(0).split("\t");
            if(strArr.length > 5 && strArr[2].equals(Constant.MATCH_TRANSACTION_TYPE_PURCHASE)){
                if(!strArr[4].equals(Constant.MATCH_RETURN_CODE)){
                    handleErrorCodeCount(strArr,resultDataBoltTime);
                }
            }
            collector.ack(input);
            collector.fail(input);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private void handleErrorCodeCount(String[] strArr,long resultDataBoltTime) {
        Long errorCodeCount = errorCodeMap.get(strArr[4]);
        if (errorCodeCount == null){
            errorCodeCount = 0L;
        }
         errorCodeCount ++ ;
         errorCodeMap.put(strArr[4],errorCodeCount);
         collector.emit(new Values(strArr[4],errorCodeCount,strArr[0],resultDataBoltTime));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("errorCode","errorCodeCount","id","resultDataBoltTime"));

    }
}